package com.wpp.dubbo.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Classname TPMonitorFilter
 * @Description
 * @Date 2021/1/20 20:50
 * @Created by wpp25
 */
@Activate(group = {CommonConstants.CONSUMER, CommonConstants.PROVIDER})
public class TPMonitorFilter implements Filter,Runnable {


    /**
     * key ： 方法名
     * value ： 方法执行的基本信息
     */
    private static final Map<String, List<QPSInfo>> cache = new ConcurrentHashMap();

    public TPMonitorFilter() {

        // 每5秒 打印 TP90和 TP99 的情况
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(this, 5, 5, TimeUnit.SECONDS);

        // 每一分钟 移出超过1分钟的统计数据
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(new RemoveExpiredData(),60,60, TimeUnit.SECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long startTime = System.currentTimeMillis();
        try {
            return invoker.invoke(invocation);
        } finally {
            invokeTime(invocation, System.currentTimeMillis() - startTime);
        }
    }

    /**
     * 记录 消费者 每次消费的数据
     * @param invocation
     * @param consumeTime
     */
    private void invokeTime(Invocation invocation, long consumeTime) {
        cache.putIfAbsent(invocation.getMethodName(), new ArrayList<>());
        List<QPSInfo> qpsInfos = cache.get(invocation.getMethodName());
        QPSInfo qpsInfo = new QPSInfo();
        qpsInfo.setExecuteTime(consumeTime);
        qpsInfo.setEndTime(System.currentTimeMillis());
        qpsInfo.setName(invocation.getMethodName());
        qpsInfos.add(qpsInfo);
    }



    /**
     * 统计并打印 tp99 和 tp90 的信息
     */
    @Override
    public void run() {
        System.out.println("start print=============================================");
        String currentTime = LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME);

        for (Map.Entry<String, List<QPSInfo>> entry : cache.entrySet()) {
            List<QPSInfo> tpRate90 = getTPRate(entry.getValue(), 0.90);

            System.out.println(currentTime + " : " + entry.getKey() + "的TP90: " + tpRate90.get(0).getExecuteTime());

            List<QPSInfo> tpRate99 = getTPRate(entry.getValue(), 0.99);
            System.out.println(currentTime + " : " + entry.getKey() + "的TP99: " + tpRate99.get(0).getExecuteTime());
            System.out.println("--------------------------------------------------------------");
        }
        System.out.println("end print================================================");

    }

    /**
     * 获取 rate 内 消费队列
     * @param qpsInfos
     * @param rate
     * @return
     */
    private List<QPSInfo> getTPRate(List<QPSInfo> qpsInfos, Double rate) {
        // 复制一个 队列 防止并发 执行 影响判断
        List<QPSInfo> copyList = new ArrayList<>(qpsInfos);
        Long endTime = System.currentTimeMillis();
        // 一分钟
        Long startTime = endTime - 60000;
        List<QPSInfo> sortedList = copyList.stream().filter(qpsInfo -> qpsInfo.getEndTime() > startTime && qpsInfo.getEndTime() < endTime)
                .sorted(Comparator.comparingLong(QPSInfo::getExecuteTime))
                .collect(Collectors.toList());
        List<QPSInfo> returnList = new ArrayList<>();
        for (int i = 0; i < sortedList.size() * rate; i++) {
            returnList.add(sortedList.get(i));
        }
        return returnList;
    }

    // 删掉 过期时间 todo
    static class RemoveExpiredData implements  Runnable {

        @Override
        public void run() {
            System.out.println("remove...");
        }
    }
}
